package org.example.rocketmq.msg_type.order;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.*;
import org.apache.rocketmq.common.message.MessageExt;
import org.example.rocketmq.NamesrvAddr;

import java.time.LocalDateTime;
import java.util.List;

public class ConsumerOrderSx1 {
    public static void main(String[] args) throws Exception {
        consumer();
    }

    /**
     * 顺序消息值生产者局部有序发送到一个 queue, 但多个 queue 之间是全局无序的。
     * 顺序消息消费者样例： 通过 MessageListenerOrderly 消费者每次读取消息都只从一个 queue 中获取（通过加锁的方式实现）
     */
    public static void consumer() throws Exception {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("consumer_order_group_01");
        consumer.setNamesrvAddr(NamesrvAddr.get());
//        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
        consumer.subscribe("TopicTest", "*");
        consumer.setConsumeThreadMin(1);
        consumer.setConsumeThreadMax(1);

        consumer.registerMessageListener(new MessageListenerOrderly() { //  顺序消费
            @Override
            public ConsumeOrderlyStatus consumeMessage(List<MessageExt> list, ConsumeOrderlyContext consumeOrderlyContext) {
                for (int i = 0; i < list.size(); i++) {
                    System.out.println(i + "_消息消费成功_" + new String(list.get(i).getBody()));

                }


                System.out.println(LocalDateTime.now());
                try {
                    Thread.sleep(1000 * 3);
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
                System.out.println("***********************************");

                return ConsumeOrderlyStatus.SUCCESS;
            }
        });


        /*consumer.registerMessageListener(new MessageListenerConcurrently() { // 并发消费
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                for (int i = 0; i < list.size(); i++) {
                    System.out.println(i + "_消息消费成功_" + new String(list.get(i).getBody()));
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });*/

        consumer.start();
        System.out.printf("Consumer started %n");
    }
}
